package com.cneport.das.server.util.amq;

import cn.hutool.core.io.IoUtil;
import com.cneport.das.common.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQMessageConsumer;
import org.apache.activemq.Message;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

import javax.jms.*;
import java.util.ArrayList;
import java.util.List;

/**
 * @author GavinZhang
 * @version 1.0
 * @desc： ActiveMQ 发送和收取的助手
 * @company xxxxx 公司
 * @date 2021/8/16 15:31
 */
@Component
@Slf4j
public class ActiveMqHelper {

    @Autowired(required = false)
    @Qualifier("amq_JmsTemplate")
    private JmsTemplate jmsTemplate;

    /**
     * 发送消息到队列
     *
     * @param jms
     * @param queue
     * @param msg
     */
    public void sendToQueue(JmsTemplate jms, String queue, String msg) {
        jms.convertAndSend(new ActiveMQQueue(queue), msg);
    }

    /**
     * 发送消息到队列
     *
     * @param queue
     * @param msg
     */
    public void sendToQueue(String queue, String msg) {
        jmsTemplate.convertAndSend(new ActiveMQQueue(queue), msg);
    }

    /**
     * 发送消息到主题
     *
     * @param jms
     * @param topic
     * @param msg
     */
    public void sendTopic(JmsTemplate jms, String topic, String msg) {
        jms.convertAndSend(new ActiveMQTopic(topic), msg);
    }

    /**
     * 发送消息到主题
     *
     * @param topic
     * @param msg
     */
    public void sendTopic(String topic, String msg) {
        jmsTemplate.convertAndSend(new ActiveMQTopic(topic), msg);
    }

    /**
     * 发送消息
     *
     * @param jms
     * @param destination
     * @param msg
     */
    public void sendMsg(JmsTemplate jms, ActiveMQDestination destination, String msg) {
        jms.convertAndSend(destination, msg);
    }

    /**
     * 发送消息
     *
     * @param destination
     * @param msg
     */
    public void sendMsg(ActiveMQDestination destination, String msg) {
        jmsTemplate.convertAndSend(destination, msg);
    }

    /**
     * 主动收取消息,一次取多条，不能大于10条，多于10，将按单条查询返回。
     *
     * @param jms
     * @param destination
     * @param count
     * @return
     */
    public String receiveMsg(JmsTemplate jms, String destination, int destinationType, int count) {
        String rtn = "";
        List list = new ArrayList();
        if (jms != null) {
            jms.setReceiveTimeout(200L);
            if (count == 0 || count == 1 || count > 10) {
                count = 1;
            }
            ConnectionFactory connectionFactory = jms.getConnectionFactory();
            Connection connection = null;
            try {
                connection = connectionFactory.createConnection();
                connection.start();
                // 创建Session
                // 有两个参数，第一个是事务，第二个是签收，后面详细介绍
                Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
                // 创建目的地（目的地有两个子接口，分别是Queue和Topic）
                // 创建消费者，指明从queue取消息
                MessageConsumer messageConsumer = null;
                if (destinationType == 0) {
                    Queue queue = session.createQueue(destination);
                    messageConsumer = session.createConsumer(queue);
                } else {
                    Topic topic = session.createTopic(destination);
                    messageConsumer = session.createConsumer(topic);
                }
                for (int i = 0; i < count; i++) {
                    Message message = (Message) messageConsumer.receive(200L);
                    if (message != null) {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            try {
                                list.add(textMessage.getText());


                            } catch (JMSException e) {
                                e.printStackTrace();
                            }
                        }
                        //确认消息
                        ((ActiveMQMessageConsumer) messageConsumer).acknowledge();
                    }

                }

            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (connection != null) {
                    IoUtil.close(connection);
                }
            }

            rtn = JsonUtil.toJsonString(list);
        }
        return rtn;


    }

    /**
     * 主动收取消息,一次取多条，不能大于10条，多于10，将按单条查询返回。
     *
     * @param destination
     * @param count
     * @return
     */
    public String receiveMsg(String destination, int destinationType, int count) {
        return receiveMsg(jmsTemplate, destination, destinationType, count);
    }

    /**
     * 主动收取消息
     *
     * @param jms
     * @param destination
     * @return
     */
    public String receiveMsg(JmsTemplate jms, String destination) {
        return receiveMsg(jms, destination, 0, 1);
    }

    /**
     * 主动收取消息
     *
     * @param destination
     * @return
     */
    public String receiveMsg(String destination) {
        return receiveMsg(jmsTemplate, destination, 0, 1);
    }

}
